package com.belf.console.service.impl;

import belf.migrate.engine.admin.taskconf.TaskConf;
import com.alibaba.fastjson2.JSONObject;
import com.belf.common.config.MigrateConfig;
import com.belf.common.core.domain.AjaxResult;
import com.belf.common.core.domain.entity.SysUser;
import com.belf.common.utils.DateUtils;
import com.belf.console.agent.migrate.BuildTaskConf;
import com.belf.console.agent.migrate.JobPush;
import com.belf.console.agent.migrate.job.SynJob;
import com.belf.console.domain.DataBatchConf;
import com.belf.console.domain.DataReportInfo;
import com.belf.console.domain.DataReportObject;
import com.belf.console.service.IDataBatchConfService;
import com.belf.console.service.IDataReportInfoService;
import com.belf.console.service.IDataReportObjectService;
import com.belf.console.domain.DataSourceInfo;
import com.belf.console.domain.DataTaskConf;
import com.belf.console.domain.DataTaskLog;
import com.belf.console.mapper.DataTaskConfMapper;
import com.belf.console.service.*;
import com.belf.quartz.service.ISysJobService;
import com.belf.system.service.ISysConfigService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

import static belf.migrate.engine.admin.taskconf.JobCode.UNKNOW_ERROR;
import static com.belf.common.constant.BelfCodeConstants.*;
import static com.belf.common.core.domain.AjaxResult.CODE_TAG;
import static com.belf.common.core.domain.AjaxResult.DATA_TAG;

/**
 * 任务配置Service业务层处理
 * 
 * @author belf
 * @date 2024-06-20
 */
@Service
public class DataTaskConfServiceImpl implements IDataTaskConfService {

    private static Logger LOGGER = LoggerFactory.getLogger(DataTaskConfServiceImpl.class.getName());
    @Autowired
    private DataTaskConfMapper dataTaskConfMapper;
    @Autowired
    private MigrateConfig migrateConfig;
    @Autowired
    private IDataTaskLogService dataTaskLogService;
    @Autowired
    private IDataSourceInfoService dataSourceInfoService;
    @Autowired
    private IDataBatchConfService dataBatchConfService;
    @Autowired
    private IDataReportInfoService dataReportInfoService;
    @Autowired
    private IDataReportObjectService dataReportObjectService;
    @Autowired
    private ISysConfigService configService;

    @Autowired
    private ISysJobService sysJobService;
    @Autowired
    private IBelfTaskJobService belfTaskJobService;
    @Autowired
    private IDataTaskJobManageService taskJobManageService;

    /**
     * 查询任务配置
     * 
     * @param id 任务配置主键
     * @return 任务配置
     */
    @Override
    public DataTaskConf selectDataTaskConfById(Long id)
    {
        DataTaskConf dataTaskConf = dataTaskConfMapper.selectDataTaskConfById(id);
        if (dataTaskConf!=null && dataTaskConf.getBatchStatus() != null && dataTaskConf.getBatchStatus() == BELF_DATA_CONF_MODULE_LINE_ON){
            Long dataTaskConfId = dataTaskConf.getId();
            List<DataBatchConf> dataBatchConfs = dataBatchConfService.selectDataBatchConfByTaskId(dataTaskConfId);
            ArrayList<DataBatchConf> jdbc = new ArrayList<>();
            ArrayList<DataBatchConf> tableBatchConfs = new ArrayList<>();
            if (dataBatchConfs!=null){
                for (DataBatchConf dataBatchConf : dataBatchConfs) {
                    if (dataBatchConf.getSelectSQL()!=null){
                        tableBatchConfs.add(dataBatchConf);
                    }else {
                        jdbc.add(dataBatchConf);
                    }
                }
            }
            dataTaskConf.setJdbc(jdbc);
            dataTaskConf.setTableBatchConf(tableBatchConfs);
        }
        return dataTaskConf;
    }

    /**
     * 查询任务配置列表
     * 
     * @param dataTaskConf 任务配置
     * @return 任务配置
     */
    @Override
    public List<DataTaskConf> selectDataTaskConfList(DataTaskConf dataTaskConf)
    {
        return dataTaskConfMapper.selectDataTaskConfList(dataTaskConf);
    }

    @Override
    public long selectDataCount(DataTaskConf dataTaskConf) {

        return dataTaskConfMapper.selectDataCount(dataTaskConf);
    }

    /**
     * 新增任务配置
     * 
     * @param task 任务配置
     * @return 结果
     */
    @Override
    public int insertDataTaskConf(DataTaskConf task)
    {
        task.setCreateTime(DateUtils.getNowDate());

        task.setUpdateTime(DateUtils.getNowDate());
        task.setTaskStatus(BELF_TASK_JOB_STATUS_INIT);

        String writebatch = configService.selectConfigByKey("data.cdc.sync.writebatch");
        task.setReaderBatch(Long.valueOf(writebatch));

        int row = dataTaskConfMapper.insertDataTaskConf(task);

        // jdbc配置保存
        if (task.getBatchStatus() != null && task.getBatchStatus() == BELF_DATA_CONF_MODULE_LINE_ON){
            List<DataBatchConf> jdbc = task.getJdbc();
            DataTaskConf finalTask1 = task;
            jdbc.forEach(dataBatchConf -> {
                dataBatchConf.setTaskId(finalTask1.getId());
                dataBatchConfService.insertDataBatchConf(dataBatchConf);
            });
        }
        List<DataBatchConf> tableBatchConf = task.getTableBatchConf();
        if (tableBatchConf!=null && tableBatchConf.size()>0){
            DataTaskConf finalTask = task;
            tableBatchConf.forEach(dataBatchConf -> {
                dataBatchConf.setTaskId(finalTask.getId());
                dataBatchConfService.insertDataBatchConf(dataBatchConf);
            });
        }


        // 调用引擎
        if (task.getStrategy() != null && task.getStrategy()==BELF_DATA_CONF_MODULE_LINE_ON) {
                executeTaskById(task.getId());
        }
        if (task.getMonitorStatus()!=null && task.getMonitorStatus()==BELF_DATA_CONF_MODULE_LINE_ON){
            task = selectDataTaskConfById(task.getId());
            belfTaskJobService.deleteTaskJob(task.getId(),task.getTaskType());
            belfTaskJobService.insertJob(task.getMonitorConf(),task.getId(),task.getTaskName(),task.getTaskType(),task.getMonitorStatus());
        }
        return row;
    }



    /**
     * 修改任务配置
     * 
     * @param task 任务配置
     * @return 结果
     */
    @Override
    public int updateDataTaskConf(DataTaskConf task)
    {
        task.setUpdateTime(DateUtils.getNowDate());
        dataBatchConfService.deleteDataBatchConfByTaskId(task.getId());
        if (task.getBatchStatus() != null && task.getBatchStatus() == BELF_DATA_CONF_MODULE_LINE_ON){
            List<DataBatchConf> jdbc = task.getJdbc();
            DataTaskConf finalTask = task;
            jdbc.forEach(dataBatchConf -> {
                dataBatchConf.setTaskId(finalTask.getId());
                dataBatchConfService.insertDataBatchConf(dataBatchConf);
            });
        }

        List<DataBatchConf> tableBatchConf = task.getTableBatchConf();
        if (tableBatchConf!=null && tableBatchConf.size()>0){
            for (DataBatchConf dataBatchConf : tableBatchConf) {
                dataBatchConf.setTaskId(task.getId());
                dataBatchConfService.insertDataBatchConf(dataBatchConf);
            }
        }

        int i = dataTaskConfMapper.updateDataTaskConf(task);
        task = selectDataTaskConfById(task.getId());
        belfTaskJobService.deleteTaskJob(task.getId(),task.getTaskType());
        // 删除旧的定时任务，添加新的定时任务
        if (task.getMonitorStatus()!=null && task.getMonitorStatus()==BELF_DATA_CONF_MODULE_LINE_ON){
            belfTaskJobService.insertJob(task.getMonitorConf(),task.getId(),task.getTaskName(),task.getTaskType(),task.getMonitorStatus());
        }
        return i;
    }

    @Override
    public int updateDataInfo(DataTaskConf task)
    {
        return dataTaskConfMapper.updateDataTaskConf(task);
    }

    public void updateData(DataTaskConf task)
    {
        task.setUpdateTime(DateUtils.getNowDate());
        dataTaskConfMapper.updateDataTaskConf(task);
    }

    /**
     * 批量删除任务配置
     * 
     * @param ids 需要删除的任务配置主键
     * @return 结果
     */
    @Override
    public int deleteDataTaskConfByIds(Long[] ids)
    {
        dataBatchConfService.deleteDataBatchConfByTaskIds(ids);
        if (ids != null) {
            for (Long taskId : ids) {
                deleteDataTaskConfById(taskId);
            }
        }
        return 1;
    }

    /**
     * 删除任务配置信息
     * 
     * @param id 任务配置主键
     * @return 结果
     */
    @Override
    public int deleteDataTaskConfById(Long id)
    {
        try {
            dataBatchConfService.deleteDataBatchConfByTaskId(id);
            DataTaskConf task = selectDataTaskConfById(id);
            belfTaskJobService.deleteTaskJob(id,task.getTaskType());
            return dataTaskConfMapper.deleteDataTaskConfById(id);
        } catch (Exception e) {
            LOGGER.error("删除定时任务失败", e);
        }
        return 0;
    }

    @Override
    public AjaxResult executeTaskById(Long id) {
        //  查询执行中任务，确认后台没有执行，再执行任务
        List<DataTaskConf> migrateTaskList = taskJobManageService.getMigrateTaskList();
        for (DataTaskConf dataTaskConf : migrateTaskList) {
            if (dataTaskConf.getId().equals(id)){
                return AjaxResult.warn("任务正在执行中，请勿重复执行！");
            }
        }
        DataTaskConf task = selectDataTaskConfById(id);
        DataTaskLog jobLog = createTaskLog(task);
        // 提交评估任务
        try {
            pushJob(task, jobLog);
            return AjaxResult.success();
        } catch (Exception e) {
            LOGGER.error("数据源信息 连接测试失败",e);
        }
        return AjaxResult.error();
    }

    @Override
    public AjaxResult schedulerTask(Long[] idArray) {
         return AjaxResult.success();
    }

    @Override
    public AjaxResult verify(DataTaskConf taskData) {
        String taskTypeScope = taskData.getTaskTypeScope();
        String name = taskData.getName();

        taskData = selectDataTaskConfById(taskData.getId());
        try {
            if (taskData != null){
                String sourceId = taskData.getSourceId();//
                String sinkId = taskData.getSinkId();
                String[] ids = {sourceId, sinkId};
                List<DataSourceInfo> dataSourceList = dataSourceInfoService.selectDataSourceInfoByIds(ids);
                taskData.setDataSourceInfoList(dataSourceList);

                TaskConf taskConf = BuildTaskConf.buildTaskConf(taskData);
                taskConf.setTaskTypeScope(taskTypeScope);
                taskConf.setName(name);
                String jsonString = JSONObject.toJSONString(taskConf);
                return JobPush.pushTask(migrateConfig.verify(), jsonString);
            }
        }catch (Exception e){
            LOGGER.error("独立校验失败！",e);
        }
        return AjaxResult.error();
    }

    @Override
    public Map<String, ? extends HashMap<String, Integer>> selectParseCount(SysUser user) {

        DataTaskConf task = new DataTaskConf();
        if (!user.isAdmin()){
            task.setCreateBy(user.getUserName());
        }
        HashMap<String,  HashMap<String, Integer>> resultMap = new HashMap<>();
        // 根据任务类型统计
        resultMap.put("dataJobTypeCount", dataTaskConfMapper.queryJobTypeCount(task));
        // 根据任务状态统计
        resultMap.put("dataTaskStatusCount",dataTaskConfMapper.queryTaskStatusCount(task));
        return resultMap;
    }

    @Override
    public DataTaskConf selectDataTaskConfByIdAndTaskType(DataTaskConf dataTaskConf) {
        return dataTaskConfMapper.selectDataTaskConfByIdAndTaskType(dataTaskConf);
    }


    private void pushJob(DataTaskConf task, DataTaskLog jobLogParam) {
        new Thread(() -> {
            DataTaskLog jobLog = jobLogParam;
            try {
                AjaxResult ajaxResult = SynJob.pushJob(migrateConfig.getPushJob(), task, jobLog);
                LOGGER.info("任务返回信息：{}", ajaxResult);
                if (ajaxResult.isSuccess()) {
                    jobLog = dataTaskLogService.selectDataTaskLogById(jobLog.getId());

                    JSONObject resultData = JSONObject.from(ajaxResult.get(DATA_TAG));
                    Long taskStatus = resultData.getLong(CODE_TAG);

                    jobLog.setTaskResult(resultData.toString());
                    if (resultData.get("errors")!=null){
                        jobLog.setProgramLog(resultData.get("errors").toString());
                    }
                    jobLog.setTaskStatus(taskStatus);
                    task.setTaskStatus(taskStatus);
                }else {
                    task.setTaskStatus((Long) ajaxResult.get(CODE_TAG));
                    jobLog.setTaskStatus((Long) ajaxResult.get(CODE_TAG));
                }
                if (ajaxResult.get("conf") != null) {
                    jobLog.setTaskConfig(ajaxResult.get("conf").toString());
                }
            } catch (Exception e) {
                LOGGER.error("任务提交失败", e);
                //  判断返回结果状态 更新数据库
                task.setTaskStatus( UNKNOW_ERROR);
                jobLog.setTaskStatus( UNKNOW_ERROR);
            }
            jobLog.setEndTime(DateUtils.getNowDate());
            jobLog.setTimeConsuming(DateUtils.timeDistance(jobLog.getEndTime(), jobLog.getStartTime()));
            // todo 分析保存迁移报告
            saveDataTaskReport(task, jobLog);
            updateData(task);
        }).start();
    }

    private void saveDataTaskReport(DataTaskConf task, DataTaskLog jobLog) {
        String taskResult = jobLog.getTaskResult();
        if (taskResult != null){
            JSONObject resultData = JSONObject.parseObject(taskResult);
            List<DataReportInfo> infoList = resultData.getList(DATA_TAG, DataReportInfo.class);

            resultData.remove(DATA_TAG);


            AtomicInteger total = new AtomicInteger();
            AtomicInteger completes = new AtomicInteger();
            if (infoList != null) {
                // 分解迁移报告
                infoList.forEach(dataReportInfo -> {

                    total.addAndGet(dataReportInfo.getTotal());

                    dataReportInfo.setTaskId(task.getId());
                    dataReportInfo.setJobId(jobLog.getId());
                    dataReportInfoService.insertDataReportInfo(dataReportInfo);
                    List<DataReportObject> successItems = dataReportInfo.getSuccessItems();
                    if (successItems != null) {

                        completes.addAndGet(successItems.size());

                        successItems.forEach(dataReportObject -> {
                            dataReportObject.setTaskId(task.getId());
                            dataReportObject.setJobId(jobLog.getId());
                            dataReportObject.setInfoId(dataReportInfo.getId());
                            dataReportObjectService.insertDataReportObject(dataReportObject);
                        });
                    }
                    List<DataReportObject> failureItems = dataReportInfo.getFailureItems();
                    if (failureItems != null) {
                        failureItems.forEach(dataReportObject -> {
                            dataReportObject.setTaskId(task.getId());
                            dataReportObject.setJobId(jobLog.getId());
                            dataReportObject.setInfoId(dataReportInfo.getId());
                            dataReportObjectService.insertDataReportObject(dataReportObject);
                        });
                    }
                });

                String completeRate = total.get() == 0 ? "100%" :String.format("%.2f", (double) completes.get() / total.get() * 100) + "%";
                resultData.put("successRate", completeRate);
                jobLog.setTaskResult(resultData.toString());
            }

        }
        dataTaskLogService.saveDataTaskLogResult(jobLog);
    }

    public DataTaskLog createTaskLog(DataTaskConf task) {
        DataTaskLog jobLog = new DataTaskLog();
        jobLog.setTaskType(task.getTaskType());
        jobLog.setTaskId(task.getId());
        jobLog.setCreateBy(task.getUpdateBy());

        jobLog.setJobType(task.getJobType());
        jobLog.setTaskStatus(BELF_TASK_JOB_STATUS_RUNING);
        jobLog.setTaskName(task.getTaskName());
        dataTaskLogService.insertDataTaskLog(jobLog);

        String sourceId = task.getSourceId();
        String sinkId = task.getSinkId();
        String[] ids = {sourceId, sinkId};
        List<DataSourceInfo> dataSourceList = dataSourceInfoService.selectDataSourceInfoByIds(ids);
        dataSourceList.forEach(dataSourceInfo -> {
            if (dataSourceInfo.getId().equals(sourceId)){
                dataSourceInfo.setDatabaseName(task.getSourceDatabase());
                dataSourceInfo.setSchemaName(task.getSourceDataSchema());
            }
            if (dataSourceInfo.getId().equals(sinkId)){
                dataSourceInfo.setDatabaseName(task.getSinkDatabase());
                dataSourceInfo.setSchemaName(task.getSinkDataSchema());
            }
        });
        task.setTaskStatus(BELF_TASK_JOB_STATUS_RUNING);
        task.setDataSourceInfoList(dataSourceList);
        task.setJobId(jobLog.getId());
        updateData(task);
        return jobLog;
    }
}
